package com.ndp.fb.rdb.mongo.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.DefaultJSONParser;
import com.alibaba.fastjson.parser.Feature;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;
import com.ndp.fb.mongo.api.IMongodbService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;
import java.util.List;

/**
 * Created by Hermen on 2015/3/9.
 */
@Service
public class MongodbServiceImpl implements IMongodbService {

    private Logger logger = LoggerFactory.getLogger(getClass());
    private Logger limitLog = LoggerFactory.getLogger("com.ndpmedia.pmd.fb.storagelimit");

    private static String[] SUM_FIELDS = {"clicks", "impressions", "results", "socialClicks", "socialImpressions", "socialSpent", "socialUniqueClicks", "socialUniqueImpressions", "spent", "uniqueClicks", "uniqueImpressions"};

    @Autowired
    private DB mongoDBClient;

    @Override
    public void batchSaveOrUpdateWithoutHistory(DB db, String tableName, String data, String[] where, boolean shouldHandleECData) {
        this.batchSaveOrUpdateUpSetMulti(db, tableName, data, where, false, null, null, shouldHandleECData, null);
    }

    @Override
    public void batchSaveOrUpdateWithHistory(DB db, String tableName, String data, String[] where, DB historyDB, String historyTableName, boolean shouldHandleECData, String... historyTableField) {
        this.batchSaveOrUpdateUpSetMulti(db, tableName, data, where, true, historyDB, historyTableName, shouldHandleECData, historyTableField);
    }

    /**
     * 更新历史记录表
     *
     * @param historyCollection
     * @param dbObject
     * @param dateUpdateTime
     */
    private void updateHistoryCollection(DBCollection historyCollection, DBObject dbObject, Date dateUpdateTime, String... otherField) {
        // 格式化数据中日期格式的字段，以便于存储在Mongo时保存的是date类型
        parseDateInDBObject(dbObject);
        // 格式化数据中字符串格式的字段，将其转换为JsonObject
        parseStringInDBObject(dbObject);

        DBObject historyCriteriaObject;
        historyCriteriaObject = new BasicDBObject();
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "clicks");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "cost");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "day");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "impressions");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "results");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "socialClicks");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "socialImpressions");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "socialSpent");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "socialUniqueClicks");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "socialUniqueImpressions");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "spent");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "uniqueClicks");
        putFieldFromSrcToDest(dbObject, historyCriteriaObject, "uniqueImpressions");
        for (int i = 0; i < otherField.length; i++) {
            putFieldFromSrcToDest(dbObject, historyCriteriaObject, otherField[i]);
        }
        DBObject historyUpdateObject = new BasicDBObject();
        historyUpdateObject.put("$set", dbObject);
        historyUpdateObject.put("$addToSet", new BasicDBObject("updateTimeList", dateUpdateTime));
        historyCollection.update(historyCriteriaObject, historyUpdateObject, true, true);
    }

    private void putFieldFromSrcToDest(DBObject srcObject, DBObject destObject, String field) {
        destObject.put(field, srcObject.get(field));
    }

    private void parseDateInDBObject(DBObject dbObject) {
        Object processStartTime = dbObject.get("processStartTime");
        if (null != processStartTime && (processStartTime instanceof Long || processStartTime instanceof String)) {
            dbObject.put("processStartTime", new Date(Long.parseLong(processStartTime.toString())));
        }

        Object processEndTime = dbObject.get("processEndTime");
        if (processEndTime != null && (processEndTime instanceof Long || processEndTime instanceof String)) {
            dbObject.put("processEndTime", new Date(Long.parseLong(processEndTime.toString())));
        }
    }

    /**
     * @param tableName
     * @param data
     * @param updateHistory
     * @param historyTableName
     * @param historyTableField
     * @discription
     * @author giser.yang
     * 创建时间 2015年7月17日 下午2:36:17
     */
    private void batchSaveOrUpdateUpSetMulti(DB db, String tableName, String data, String[] where, boolean updateHistory, DB historyDB, String historyTableName, boolean shouldHandleECData, String... historyTableField) {
        long start1 = System.currentTimeMillis();
        List<DBObject> tmpList = (List<DBObject>) com.mongodb.util.JSON.parse(data);
        logger.info("Json 转化耗时 >>>>> " + (System.currentTimeMillis() - start1));

        DBObject criteria = null;
        DBObject updateData = null;
        long start = System.currentTimeMillis();
        DBCollection dbCollection = db.getCollection(tableName);
        DBCollection historyCollection = null;
        if (updateHistory) {
            // 更新历史表
            historyCollection = historyDB.getCollection(historyTableName);
        }
        for (DBObject dbObject : tmpList) {
            try {
                // 格式化数据中日期格式的字段，以便于存储在Mongo时保存的是date类型
                parseDateInDBObject(dbObject);
                // 格式化数据中字符串格式的字段，将其转换为JsonObject
                parseStringInDBObject(dbObject);
                // 如果需要处理电商相关的数据，则进行电商相关数据转换
                if (shouldHandleECData) {
                    convertECData(dbObject);
                }
                // 查询条件
                criteria = new BasicDBObject();
                for (String field : where) {
                    Object value = dbObject.get(field);
                    if (null != value) {
                        criteria.put(field, value);
                    }
                }
                // 更新内容
                updateData = new BasicDBObject();
                updateData.put("$set", dbObject);
                Date dateUpdateTime = new Date();
                dbObject.put("dataUpdateTime", dateUpdateTime);
                dbCollection.update(criteria, updateData, true, true);
                if (updateHistory) {
                    // 更新历史表
                    updateHistoryCollection(historyCollection, dbObject, dateUpdateTime, historyTableField);
                }
            } catch (Exception e) {
                logger.error("batchSaveOrUpdateUpSetMulti 插入数据失败 >>>>> dbObject={}", dbObject, e);
            }
        }
        limitLog.info("插入表" + tableName + "完成,total = >>>>> " + tmpList.size());
        limitLog.info("插入完成耗时>>>>> " + (System.currentTimeMillis() - start));
    }


    private void convertECData(DBObject dbObject) {
        // 将数组类型的数据转换为对象保存
        DBObject convertArray = convertArray(dbObject);
        if (!convertArray.toMap().isEmpty()) {
            dbObject.put("convertedArray", convertArray);
        }

        // 设备转换
        if (null != dbObject.get("impression_device")) {
            String convertedDevice = convertDevice(dbObject);
            if (StringUtils.isNotEmpty(convertedDevice)) {
                dbObject.put("convertedDevice", convertedDevice);
            }
        }
    }

    /**
     * 转换设备
     *
     * @param dbObject
     * @return
     */
    private String convertDevice(DBObject dbObject) {
        String impressionDevice = dbObject.get("impression_device").toString();
        // 移动设备
        if (impressionDevice.toLowerCase().contains("ipod") || impressionDevice.toLowerCase().contains("phone")) {
            return "mobile";
        } else if (impressionDevice.toLowerCase().contains("pad") || impressionDevice.toLowerCase().contains("tablet")) { // 平板
            return "tablet";
        } else {    // 其它类型（台式电脑或者Other）
            return impressionDevice;
        }
    }

    private DBObject convertArray(DBObject dbObject) {
        DBObject result = new BasicDBObject();
        dbObject.keySet().forEach(key -> {
            Object value = dbObject.get(key);
            if (value instanceof JSONArray) {
                ((JSONArray) value).forEach(value1 -> {
                    if (value1 instanceof JSONObject) {
                        JSONObject tmpValue = (JSONObject) value1;
                        if (tmpValue.containsKey("action_type")) {
                            String action_type = tmpValue.get("action_type").toString();
                            String[] split = action_type.split("\\.");
                            convertArrayRec(split, result, 0, tmpValue, key);
                        }
                    }
                });
            }
        });
        return result;
    }

    private void convertArrayRec(String[] split, DBObject dbObject, int i, Object value, String key) {
        if (split.length - 1 != i) {
            DBObject curDBObject = null;
            if (null == dbObject.get(split[i])) {
                curDBObject = new BasicDBObject();
                dbObject.put(split[i], curDBObject);
            } else {
                curDBObject = (DBObject) dbObject.get(split[i]);
            }
            convertArrayRec(split, curDBObject, i + 1, value, key);
        } else {
            Object o = dbObject.get(split[i]);
            if (null == o) {
                dbObject.put(split[i], new BasicDBObject());
            }
            DBObject secondDBObject = (DBObject) dbObject.get(split[i]);
            if (null == secondDBObject.get(key)) {
                secondDBObject.put(key, value);
            } else {
                secondDBObject.put(key, value);
            }
        }
    }

    private void parseStringInDBObject(DBObject dbObject) {
        Object actions = dbObject.get("actions");
        if (null != actions && actions instanceof String) {
            try {
                dbObject.put("actions", JSON.parse(actions.toString()));
            } catch (Exception e) {
                logger.error("JSON转换失败，将保存原始值{}", actions);
                dbObject.put("error_actions", actions);
            }
        }

        Object uniqueActions = dbObject.get("uniqueActions");
        if (null != uniqueActions && uniqueActions instanceof String) {
            try {
                dbObject.put("uniqueActions", JSON.parse(uniqueActions.toString()));
            } catch (Exception e) {
                logger.error("JSON转换失败，将保存原始值{}", uniqueActions);
                dbObject.put("error_uniqueActions", uniqueActions);
            }
        }

        convertBigDecimal2Double(dbObject, "actionValues");
        convertBigDecimal2Double(dbObject, "costPerActionType");
        convertBigDecimal2Double(dbObject, "costPerUniqueActionType");

        Object relevanceScore = dbObject.get("relevanceScore");
        if (null != relevanceScore && relevanceScore instanceof String) {
            try {
                dbObject.put("relevanceScore", JSON.parse(relevanceScore.toString()));
            } catch (Exception e) {
                logger.error("JSON转换失败，将保存原始值{}", relevanceScore);
                dbObject.put("error_relevanceScore", relevanceScore);
            }
        }

        Object websiteCtr = dbObject.get("websiteCtr");
        if (null != websiteCtr && websiteCtr instanceof String) {
            try {
                DefaultJSONParser defaultJSONParser = new DefaultJSONParser(websiteCtr.toString());
                defaultJSONParser.config(Feature.UseBigDecimal, false);
                dbObject.put("websiteCtr", defaultJSONParser.parse());
            } catch (Exception e) {
                logger.error("JSON转换失败，将保存原始值{}", websiteCtr);
                dbObject.put("error_websiteCtr", websiteCtr);
            }
        }
    }

    private void convertBigDecimal2Double(DBObject dbObject, String key) {
        Object srcDBObject = dbObject.get(key);
        if (null != srcDBObject && srcDBObject instanceof String) {
            try {
                JSONArray jsonArray = JSON.parseArray(srcDBObject.toString());
                for (int i = 0; i < jsonArray.size(); i++) {
                    JSONObject jsonObject = jsonArray.getJSONObject(i);
                    double value = jsonObject.getDoubleValue("value");
                    jsonObject.put("value", value * 100);
                    if (null != jsonObject.get("1d_click")) {
                        jsonObject.put("1d_click", jsonObject.getDoubleValue("1d_click") * 100);
                    }
                }
                dbObject.put(key, jsonArray);
            } catch (Exception e) {
                logger.error("JSON转换失败，将保存原始值{}", srcDBObject);
                dbObject.put("error_" + key, srcDBObject);
            }
        }
    }

    /**
     * @param tableName
     * @param data
     * @param historyTableField
     * @discription
     * @author giser.yang
     * 创建时间 2015年7月23日 上午11:43:20
     */

    @Override
    public void updateHistory(DB historyDB, String tableName, String data, String... historyTableField) {
        long start1 = System.currentTimeMillis();
        List<DBObject> tmpList = (List<DBObject>) com.mongodb.util.JSON.parse(data);
        logger.info("Json 转化耗时 >>>>> " + (System.currentTimeMillis() - start1));
        DBCollection historyCollection = historyDB.getCollection(tableName);
        for (DBObject dbObject : tmpList) {
            try {
                updateHistoryCollection(historyCollection, dbObject, new Date(), historyTableField);
            } catch (Exception e) {
                logger.error("updateHistory 插入数据失败 >>>>> tableName={}, data={}, historyTableField={}", tableName, data, historyTableField, e);
            }
        }
        limitLog.info("插入表" + tableName + "完成,total = >>>>> " + tmpList.size());
        limitLog.info("插入完成耗时>>>>> " + (System.currentTimeMillis() - start1));
    }


    @Override
    public void batchSaveOrIncreaseStatsData(DB db, String tableName, String data, String[] where) {
        long start1 = System.currentTimeMillis();
        List<DBObject> tmpList = (List<DBObject>) com.mongodb.util.JSON.parse(data);
        logger.info("Json 转化耗时 >>>>> " + (System.currentTimeMillis() - start1));
        DBObject criteria = null;
        DBObject updateData = null;
        DBObject incChildData = null;
        long start = System.currentTimeMillis();
        DBCollection dbCollection = db.getCollection(tableName);
        for (DBObject dbObject : tmpList) {
            try {
                // 格式化数据中日期格式的字段，以便于存储在Mongo时保存的是date类型
                parseDateInDBObject(dbObject);
                // 格式化数据中字符串格式的字段，将其转换为JsonObject
                parseStringInDBObject(dbObject);
                // 查询条件
                criteria = new BasicDBObject();
                for (String field : where) {
                    Object value = dbObject.get(field);
                    if (null != value) {
                        criteria.put(field, value);
                    }
                }
                incChildData = new BasicDBObject();
                for (String sumField : SUM_FIELDS) {
                    Object value = dbObject.get(sumField);
                    if (null != value) {
                        incChildData.put(sumField, value);
                        dbObject.removeField(sumField);
                    }
                }
                // 由于cost不能进行相加运算，因此不再保存cost字段，需要时实时计算
                dbObject.removeField("cost");
                // 更新内容
                updateData = new BasicDBObject();
                updateData.put("$set", dbObject);
                updateData.put("$inc", incChildData);
                Date dateUpdateTime = new Date();
                dbObject.put("dataUpdateTime", dateUpdateTime);
                dbCollection.update(criteria, updateData, true, true);
            } catch (Exception e) {
                logger.error("batchSaveOrIncreaseStatsData 插入/更新数据失败 >>>>> dbObject={}", dbObject, e);
            }
        }
        limitLog.info("插入/更新表" + tableName + "完成,total = >>>>> " + tmpList.size());
        limitLog.info("插入/更新完成耗时>>>>> " + (System.currentTimeMillis() - start));
    }

}
